package producer;

import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.ProducerConfig;
import org.apache.kafka.clients.producer.ProducerRecord;

import java.text.SimpleDateFormat;
import java.util.Date;
import java.util.Properties;
import java.util.Random;

public class kfkproducer {
    private static final Random random = new Random();

    public static void main(String[] args) {
        // 1. 创建 kafka 生产者的配置对象
        Properties properties = new Properties();
        // 2. 给 kafka 配置对象添加配置信息：bootstrap.servers
        properties.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "192.168.244.11:9092");
        properties.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringSerializer");
        properties.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringSerializer");


        KafkaProducer<String, String> kafkaProducer = new KafkaProducer<String, String>(properties);
        while (true) {
            while (true) {
                String data = generateRandomData();
                kafkaProducer.send(new ProducerRecord<String, String>("lol", data));
                System.out.println("Sent data: " + data);

                try {
                    Thread.sleep(5000); // 每隔5秒推送一次
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            }
        }


    }

    private static String generateRandomData() {
        String[] names = {"Alice", "Bob", "Charlie", "David", "Eva"};
        String[] classIds = {"101", "102", "103", "104"};

        String classId = classIds[random.nextInt(classIds.length)];
        String name = names[random.nextInt(names.length)];
        int gender = random.nextBoolean() ? 1 : 0;  // 1为男，0为女
        String birthDate = generateRandomDate1();
        String studentId = generateRandomStudentId();
        int semester = random.nextInt(8) + 1;  // 学期1~8
        String enrollmentStatus = random.nextBoolean() ? "A" : "L"; // A在校, L毕业

        return String.format("%s\t%s\t%d\t%s\t%s\t%d\t%s",
                classId, name, gender, birthDate, studentId, semester, enrollmentStatus);
    }

    private static String generateRandomDate1() {
        SimpleDateFormat sdf = new SimpleDateFormat("yyyy-MM-dd");
        long minDate = 315569520000L; // 1970-01-01
        long maxDate = System.currentTimeMillis(); // 当前时间
        long randomDate = minDate + (long) (Math.random() * (maxDate - minDate));
        return sdf.format(new Date(randomDate));
    }

    private static String generateRandomStudentId() {
        return String.format("%08d", random.nextInt(100000000));
    }
}
